package org.btik.video.server.device.udp2;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.btik.context.thread.SystemControlAble;
import org.btik.light.tool.thread.NamePrefixThreadFactory;
import org.btik.video.server.VideoServer;
import org.btik.video.server.iface.DevChannel;
import org.btik.video.server.iface.SoSendCheckTarget;
import org.btik.video.server.iface.VideoChannel;
import org.btik.video.server.push.VideoChannelFactory;
import org.btik.video.server.push.web.SoSendTimeoutTicker;

import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.*;

public class NewUDPDeviceChannel extends Thread implements DevChannel, SystemControlAble {

    private final Log log = LogFactory.getLog(NewUDPDeviceChannel.class);

    public static final int SN_LEN = 12;
    private volatile boolean runFlag = true;

    /**
     * 帧通道端口号
     */
    private int streamPort;

    private VideoChannelFactory videoChannelFactory;

    private ExecutorService executorService;

    private BufferPool bufferPool;


    private FrameDispatcher[] frameDispatchers;

    /**
     * 帧分发线程数量，随着设备增多可以适当增加
     */
    private int dispatcherPoolSize = 8;
    private int maxDispatcherPoolIndex;

    private int videoChannelCount = 128;

    /**
     * 防止接收图片过快而，浏览器接收过慢，导致帧积压，当帧接收时间到转发时间超过以下延时之后将会被丢弃。
     */
    private long frameDelayTimeout = 300;

    public void setStreamPort(int streamPort) {
        this.streamPort = streamPort;
    }

    public void setBufferPool(BufferPool bufferPool) {
        this.bufferPool = bufferPool;
    }

    public void setVideoChannelFactory(VideoChannelFactory videoChannelFactory) {
        this.videoChannelFactory = videoChannelFactory;
    }

    public void setDispatcherPoolSize(int dispatcherPoolSize) {
        // 至少一个线程
        this.dispatcherPoolSize = Math.max(1, dispatcherPoolSize);
        this.maxDispatcherPoolIndex = this.dispatcherPoolSize - 1;
    }


    public void setVideoChannelCount(int videoChannelCount) {
        this.videoChannelCount = videoChannelCount;
    }

    public void setFrameDelayTimeout(long frameDelayTimeout) {
        this.frameDelayTimeout = frameDelayTimeout;
    }

    private VideoChannel[] videoChannelTable;

    @Override
    public synchronized void start() {
        log.info("init buffer pool");

        log.info("start dispatchers");
        frameDispatchers = new FrameDispatcher[dispatcherPoolSize];
        videoChannelTable = new VideoChannel[videoChannelCount];
        executorService = new ThreadPoolExecutor(dispatcherPoolSize, dispatcherPoolSize,
                0, TimeUnit.SECONDS, new LinkedBlockingQueue<>(), new NamePrefixThreadFactory("frameDispatcher"));
        int bufferPoolSize = bufferPool.getBufferPoolSize();
        for (int i = 0; i < dispatcherPoolSize; i++) {
            FrameDispatcher msgDispatcher = new FrameDispatcher(bufferPoolSize / dispatcherPoolSize + 1);
            frameDispatchers[i] = msgDispatcher;
            executorService.submit(msgDispatcher);
        }

        log.info("udp channel loaded");
        super.start();
    }

    @Override
    public void shutdown(String msg) {
        runFlag = false;
        // 无消息导致阻塞时，没有读到flag,帮助退出阻塞
        for (FrameDispatcher frameDispatcher : frameDispatchers) {
            frameDispatcher.messages.add(new FrameBuffer(new byte[0]));
        }
        // 线程池核心线程也需要停止
        executorService.shutdown();
    }


    @Override
    public void run() {

        try (DatagramSocket serverSocket = new DatagramSocket(streamPort)) {
            FrameBuffer frameBuffer = bufferPool.getFrameBuffer();
            DatagramPacket datagramPacket = new DatagramPacket(frameBuffer.data, 0, frameBuffer.data.length);
            while (runFlag) {
                serverSocket.receive(datagramPacket);
                frameBuffer.size = datagramPacket.getLength() - 1;
                frameBuffer.channelIndex = frameBuffer.data[frameBuffer.size] & 0xff;
                frameBuffer.time = System.currentTimeMillis();
                frameDispatchers[(frameBuffer.channelIndex % maxDispatcherPoolIndex)].messages.add(frameBuffer);
                // 切换缓冲区
                frameBuffer = bufferPool.getFrameBuffer();
                datagramPacket.setData(frameBuffer.data);
            }
        } catch (IOException e) {
            log.info(" start server failed:" + e.getMessage());
        }
    }


    @Override
    public int channelIdLen() {
        return SN_LEN;
    }

    class FrameDispatcher implements Runnable {
        ArrayBlockingQueue<FrameBuffer> messages;

        @Override
        public void run() {

            try {
                while (runFlag) {
                    FrameBuffer segment = messages.take();
                    try {
                        int channelIndex = segment.channelIndex;
                        if (channelIndex < 0 || channelIndex > videoChannelCount) {
                            if (log.isTraceEnabled()) {
                                log.trace("The channel index is illegal");
                            }
                            continue;
                        }
                        VideoChannel videoChannel = videoChannelTable[channelIndex];
                        if (videoChannel == null) {
                            if (log.isTraceEnabled()) {
                                log.trace("videoChannel index is not registered and may need to be restarted the ESP");
                            }
                            continue;
                        }
                        // 延时跳帧
                        if (System.currentTimeMillis() > segment.time + frameDelayTimeout) {
                            if (log.isTraceEnabled()) {
                                log.trace("frame time out continue");
                            }
                            continue;
                        }
                        videoChannel.sendFrame(segment.data, segment.size);
                    } catch (Exception e) {
                        if (runFlag) {
                            log.info("", e);
                        } else {
                            break;
                        }
                    } finally {
                        // 归还到池里
                        bufferPool.returnBuffer(segment);
                    }

                }
            } catch (InterruptedException e) {
                e.printStackTrace();
                log.info("exit by:" + e);
            }
            log.info("exit : " + getName());
        }

        public FrameDispatcher(int size) {
            messages = new ArrayBlockingQueue<>(size);
        }
    }

    /**
     * 字符串sn本身是mac地址十六进制的值，直接获取编码便是十六进制字符
     */
    public void attachMjpegVideoChannel(int channelIndex, String sn) {
        videoChannelTable[channelIndex] = videoChannelFactory.createMJPEGVideoChannel(channelIndex, sn);
    }

    @Override
    public void attachRawUDPVideoChannel(int channelIndex, String sn) {
        videoChannelTable[channelIndex] = videoChannelFactory.createRawUDPVideoChannel(channelIndex, sn);
    }
}
